package com.sh.data.engine.web.webapi.controller.integration;

import com.sh.data.engine.application.model.dto.base.PagerResponse;
import com.sh.data.engine.application.model.dto.common.DirectoryResponse;
import com.sh.data.engine.application.model.dto.integration.*;
import com.sh.data.engine.application.model.request.common.UpsertDirectoryRequest;
import com.sh.data.engine.application.model.request.integration.DataSourceSaveRequest;
import com.sh.data.engine.application.model.request.integration.FlinkDataSyncTaskRequest;
import com.sh.data.engine.application.model.request.integration.PreviewDataForMqRequest;
import com.sh.data.engine.application.model.request.integration.PreviewJsonDataForSpRequest;
import com.sh.data.engine.application.service.integration.FlinkDataSyncApplicationService;
import com.sh.data.engine.common.annotation.fuzzyQuery.FuzzyMethod;
import com.sh.data.engine.common.enumDefinition.ProjectStatisticTypeEnum;
import com.sh.data.engine.domain.annotation.LogExecution;
import com.sh.data.engine.domain.annotation.ProjectStatisticExecution;
import com.sh.data.engine.domain.annotation.userlog.Business;
import com.sh.data.engine.domain.annotation.userlog.EngineOperation;
import com.sh.data.engine.domain.datadev.offline.model.domain.OnlineTaskLogDomain;
import com.sh.data.engine.domain.integration.online.service.FlinkDataSyncService;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.web.webapi.controller.base.response.RestResponseEntity;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.media.Schema;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.format.annotation.DateTimeFormat;
import org.springframework.web.bind.annotation.*;

import java.util.Date;
import java.util.List;
import java.util.Set;

/**
 * @author: mengzheng.mhc
 * @date: 2024/7/2 15:17
 */
@RestController
@RequestMapping("/integration/online")
@Slf4j
@Tag(name = "数据集成-实时同步", description = "数据集成-实时同步API")
public class FlinkDataSyncTaskController {

    @Autowired
    private FlinkDataSyncService flinkDataSyncService;

    @Autowired
    private FlinkDataSyncApplicationService flinkDataSyncApplicationService;

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 1, isAdd = 1)
    @Operation(summary = "保存实时采集任务", description = "保存实时采集任务")
    @PostMapping("/insert")
    @ProjectStatisticExecution(type = ProjectStatisticTypeEnum.INTEGRATE_TASK_ONLINE)
    @LogExecution(operation = EngineOperation.ADD_UPDATE, business = Business.实时采集任务)
//    @ProjectStatisticExecution(description = "数据集成-离线", primaryClassification = 3,secondaryClassification = 2)
//    @Log(
//        business = Business.实时采集任务,
//        operation = Operation.ADD_UPDATE)
    public RestResponseEntity<Boolean> insert(@RequestBody FlinkDataSyncTaskRequest request) {
        var result = flinkDataSyncApplicationService.insert(request);
        return RestResponseEntity.success(result);
    }

    @SneakyThrows
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 1, isAdd = 1)
    @Operation(summary = "发布任务", description = "发布任务")
//    @Log(
//        business = Business.实时采集任务,
//        operation = Operation.PUBLISH)
    @GetMapping("/publish")
    @LogExecution(operation = EngineOperation.PUBLISH, business = Business.实时采集任务)
    public RestResponseEntity<Boolean> publish(
        @Parameter(description = "实时采集任务id") @RequestParam(value = "id") Long id) {
        var result = flinkDataSyncApplicationService.publish(id);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/task/log")
    @Operation(summary = "任务日志")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0)
    public RestResponseEntity<OnlineTaskLogResponse> taskLog(
        @Parameter(description = "记录Id", required = true) @RequestParam("recordId") Long jobId,
        @Parameter(description = "偏移量", required = true) @RequestParam("offset") Integer offset) {

        OnlineTaskLogDomain taskLog = flinkDataSyncService.getTaskLog(jobId, offset);

        OnlineTaskLogResponse onlineTaskLogResponse =
            ConvertUtil.copyProperties(taskLog, OnlineTaskLogResponse.class);
        return new RestResponseEntity<>(onlineTaskLogResponse);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 1, isAdd = 1)
    @Operation(summary = "重跑任务", description = "重跑任务")
    @GetMapping("/reRun")
    @LogExecution(operation = EngineOperation.START_TASK, business = Business.实时采集任务)
//    @Log(
//        business = Business.实时采集任务,
//        operation = Operation.START_TASK)
    public RestResponseEntity<Boolean> reRun(
        @Parameter(description = "实时采集任务id") @RequestParam(value = "id") Long id) {
        var result = flinkDataSyncApplicationService.reRun(id);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 1, isAdd = 1)
    @Operation(summary = "下线任务", description = "下线任务")
    @GetMapping("/unPublish")
    @LogExecution(operation = EngineOperation.DEPUBLISH, business = Business.实时采集任务)
//    @Log(
//        business = Business.实时采集任务,
//        operation = Operation.DEPUBLISH)
    public RestResponseEntity<Boolean> unPublish(
        @Parameter(description = "实时采集任务id") @RequestParam(value = "id") Long id) {
        var result = flinkDataSyncApplicationService.unPublish(id);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 1, isAdd = 1)
    @Operation(summary = "更新实时采集任务", description = "更新实时采集任务")
    @PostMapping("/update")
    @LogExecution(operation = EngineOperation.UPDATE, business = Business.实时采集任务)
//    @Log(
//        business = Business.实时采集任务,
//        operation = Operation.UPDATE)
    public RestResponseEntity<Boolean> update(@RequestBody FlinkDataSyncTaskRequest request) {
        var result = flinkDataSyncApplicationService.update(request);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 1, isAdd = 1)
    @Operation(summary = "删除实时采集任务", description = "删除实时采集任务")
    @GetMapping("/delete")
    @LogExecution(operation = EngineOperation.DEL, business = Business.实时采集任务)
//    @Log(
//        business = Business.实时采集任务,
//        operation = Operation.DEL)
    @ProjectStatisticExecution(type = ProjectStatisticTypeEnum.INTEGRATE_TASK_ONLINE)
//    @ProjectStatisticExecution(description = "数据集成-实时", primaryClassification = 3,secondaryClassification = 2)
    public RestResponseEntity<Boolean> delete(
        @Parameter(description = "实时采集任务id") @RequestParam(value = "id") Long id) {
        var result = flinkDataSyncApplicationService.delete(id);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0, isAdd = 1)
    @GetMapping("/getDatabasesByDsType")
    @Operation(summary = "获取数据源")
    public RestResponseEntity<List<FlinkDatabaseResponse>> getDatabasesByDsType(
        @Parameter(description = "数据源来源 0-来源 1-目标", required = true) @RequestParam(value = "direction")
        Integer direction,
        @Parameter(description = "来源表数据类型", required = false)
        @RequestParam(value = "dsType", required = false)
        String dsType) {
        var result = flinkDataSyncApplicationService.getDataBaseByDirection(direction, dsType);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getDatabasesForSp")
    @Operation(summary = "获取数据源（for算盘）")
    public RestResponseEntity<List<FlinkDatabaseResponse>> getDatabasesForSp(
        @Parameter(description = "项目id", required = true) @RequestParam(value = "projectId")
        Long projectId) {
        var result = flinkDataSyncApplicationService.getListByDsTypes(projectId);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0, isAdd = 1)
    @GetMapping("/getTables")
    @Operation(summary = "获取数据源的表列表")
    public RestResponseEntity<List<FlinkTableResponse>> getTables(
        @Parameter(description = "数据源id", required = true) @RequestParam Long dsId,
        @Parameter(description = "schema") @RequestParam(value = "schema", required = false) String schema,
        @Parameter(description = "数据源来源 0-来源 1-目标", required = true) @RequestParam(value = "direction") Integer direction) {
        var result = flinkDataSyncApplicationService.getTables(dsId, schema, direction);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0, isAdd = 1)
    @GetMapping("/getTableFields")
    @Operation(summary = "获取数据源的字段列表")
    public RestResponseEntity<List<FlinkTableFieldResponse>> getTableFields(
        @Parameter(description = "数据源id", required = true) @RequestParam Long dsId,
        @Parameter(description = "表名", required = true) @RequestParam String tableName) {
        var result = flinkDataSyncApplicationService.getTableFields(dsId, tableName);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0, isAdd = 1)
    @GetMapping("/getTopic")
    @Operation(summary = "获取kafka类型数据源的topic信息")
    public RestResponseEntity<Set<String>> getTopics(
        @Parameter(description = "数据源id", required = true) @RequestParam Long dsId) {
        var result = flinkDataSyncApplicationService.getTopics(dsId);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0, isAdd = 1)
    @GetMapping("/page")
    @Operation(summary = "实时采集任务列表")
    public RestResponseEntity<PagerResponse<FlinkDataSyncTaskPageResponse>> getFlinkTaskPage(
        @Parameter(description = "当前页", schema = @Schema(defaultValue = "1"))
        @RequestParam(required = false, defaultValue = "1")
        Integer pageNum,
        @Parameter(description = "每页大小", schema = @Schema(defaultValue = "10"))
        @RequestParam(required = false, defaultValue = "10")
        Integer pageSize,
        @Parameter(description = "任务名称") @RequestParam(required = false) String taskName,
        @Parameter(description = "任务状态") @RequestParam(required = false) Integer taskStatus,
        @Parameter(description = "目录id") @RequestParam(required = false) Long dirId,
        @Parameter(description = "运行状态") @RequestParam(required = false) Integer taskRunStatus,
        @Parameter(description = "开始时间（发布时间）")
        @RequestParam(required = false)
        @DateTimeFormat(pattern = "yyyy-MM-dd")
        Date startTime,
        @Parameter(description = "结束时间（发布时间）")
        @RequestParam(required = false)
        @DateTimeFormat(pattern = "yyyy-MM-dd")
        Date endTime) {
        var result = flinkDataSyncApplicationService.getFlinkTaskPage(
            pageNum, pageSize, taskName, taskStatus, dirId, taskRunStatus, startTime, endTime);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0, isAdd = 1)
    @GetMapping("/getSummary")
    @Operation(summary = "运行概览")
    public RestResponseEntity<FlinkTaskSummaryResponse> getSummary() {
        var result = flinkDataSyncApplicationService.getSummary();
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0, isAdd = 1)
    @Operation(summary = "预览数据，返回最多10行记录", description = "预览数据，返回最多10行记录")
    @GetMapping(value = "/previewData")
    public RestResponseEntity<OnlinePreviewDataResponse> previewData(
        @Parameter(description = "数据源id") @RequestParam(value = "dsId") Long dsId,
        @Parameter(description = "表名") @RequestParam(value = "tableName") String tableName) {
        var result = flinkDataSyncApplicationService.previewData(dsId, tableName);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0, isAdd = 1)
    @Operation(summary = "获取建表sql")
    @GetMapping(value = "/getCreateTableSql")
    public RestResponseEntity<String> createTargetTable(
        @Parameter(description = "来源数据源id") @RequestParam(value = "dsId") Long dsId,
        @Parameter(description = "来源数据表名称") @RequestParam(value = "tableName") String tableName,
        @Parameter(description = "目标数据源id") @RequestParam(value = "targetDsId") Long targetDsId,
        @Parameter(description = "schema") @RequestParam(value = "schema", required = false) String schema,
        @Parameter(description = "infoCase")
        @RequestParam(value = "infoCase", required = false, defaultValue = "1")
        int infoCase) {
        var result = flinkDataSyncApplicationService.createTargetTable(dsId, tableName, targetDsId, schema, infoCase);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0, isAdd = 1)
    @Operation(summary = "获取kafka/mqtt数据预览")
    @PostMapping("/previewDataForMq")
    public RestResponseEntity<List<List>> previewDataForMq(
        @RequestBody PreviewDataForMqRequest request) {
        var result = flinkDataSyncApplicationService.previewDataForMq(request);
        return RestResponseEntity.success(result);
    }

    @Operation(summary = "获取kafka/mqtt数据预览（for算盘）")
    @PostMapping("/previewJsonDataForSp")
    public RestResponseEntity<List<List>> previewJsonDataForSp(
        @RequestBody PreviewJsonDataForSpRequest request) {
        var result = flinkDataSyncApplicationService.previewJsonDataForSp(request);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 1, isAdd = 1)
    @Operation(summary = "执行建表sql")
    @PostMapping(value = "/executeCreateTableSql")
    @LogExecution(operation = EngineOperation.DATA_SYNC_CREATE_TABLE, business = Business.实时采集任务)
    @ProjectStatisticExecution(type = ProjectStatisticTypeEnum.TABLE)
//    @ProjectStatisticExecution(description = "已创建数据表", primaryClassification = 5)
//    @Log(
//        business = Business.实时采集任务,
//        operation = Operation.DATA_SYNC_CREATE_TABLE)
    public RestResponseEntity<OfflineSyncCreateTableResponse> executeCreateTableSql(
        @Parameter(description = "目标数据源id") @RequestParam(value = "dsId") Long dsId,
        @Parameter(description = "SQL脚本", required = true) @RequestParam(value = "scripts") String scripts,
        @Parameter(description = "密级", required = false) @RequestParam(value = "dataSecretId", required = false) Long dataSecretId) {
        var result = flinkDataSyncApplicationService.executeCreateTableSql(dsId, scripts, dataSecretId);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0, isAdd = 1)
    @Operation(summary = "获取任务详情")
    @GetMapping("/getDetailInfo")
    public RestResponseEntity<FlinkTaskDetailInfoResponse> getDetailInfo(
        @Parameter(description = "任务id", required = true) @RequestParam(value = "id") Long id) {
        var result = flinkDataSyncApplicationService.getDetailInfo(id);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0)
    @Operation(summary = "获取目录", description = "获取目录")
    @FuzzyMethod(fuzzyParamIndex = 0)
    @GetMapping("/getDirList")
    public RestResponseEntity<List<DirectoryResponse>> getDirList(
        @Parameter(description = "目录搜索", required = false) @RequestParam(value = "search", required = false)
        String search) {
        var result = flinkDataSyncApplicationService.getDirList(search);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 1)
    @Operation(summary = "新增、更新目录", description = "新增、更新目录")
    @PostMapping("/upsetDir")
    @LogExecution(operation = EngineOperation.ADD_UPDATE, business = Business.实时采集任务目录)
//    @Log(
//        business = Business.实时采集任务目录,
//        operation = Operation.ADD_UPDATE)
    public RestResponseEntity<Boolean> upsetDir(@RequestBody UpsertDirectoryRequest request) {
        var result = flinkDataSyncApplicationService.upsetDir(request);
        return RestResponseEntity.success(result);
    }

    //    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 1)
    @Operation(summary = "删除目录", description = "删除目录")
    @GetMapping("/deleteDir")
    @LogExecution(operation = EngineOperation.DEL, business = Business.实时采集任务目录)
//    @Log(
//        business = Business.实时采集任务目录,
//        operation = Operation.DEL)
    public RestResponseEntity<Boolean> deleteDir(Long id) {
        var result = flinkDataSyncApplicationService.deleteDir(id);
        return RestResponseEntity.success(result);
    }

    @Operation(summary = "查询数据库下所有schema", description = "查询数据库下所有schema")
    @GetMapping("/schema")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSTB, readWriteStatus = 0)
    public RestResponseEntity<List<String>> getSchemas(
        @Parameter(description = "数据源id") @RequestParam(value = "dsId") Long dsId,
        @Parameter(description = "数据源类型", required = true) @RequestParam(value = "dsType") String dsType) {
        var result = flinkDataSyncApplicationService.getSchemas(dsId, dsType);
        return RestResponseEntity.success(result);
    }

    @Operation(summary = "查询是否有kafka数据源(for算盘)", description = "查询是否有kafka数据源(for算盘)")
    @PostMapping("/checkSpKafka")
    public RestResponseEntity<Long> checkSpKafka(
        @RequestBody DataSourceSaveRequest dataSourceSaveRequest) {
        var result = flinkDataSyncApplicationService.checkSpKafka(dataSourceSaveRequest);
        return RestResponseEntity.success(result);
    }
}
